%% Author: vivimice
%% Created: Feb 7, 2012
%% Description: TODO: Add description to blocking_queue
-module(blocking_binary_queue).

%%
%% Include files
%%

%%
%% Exported Functions
%%
-export([new/0, init/0, loop/2, close/1, put/2, take/2, peek/2]).

%%
%% API Functions
%%
new() ->
	spawn_link(?MODULE, init, []).

init() ->
	loop(<<>>, []).

loop(Buffer, Observers) ->
	receive
		{From, put, Data} ->
			{NewBuffer, NewObservers} = notify_observers(<<Buffer/binary, Data/binary>>, Observers),
			From ! {self(), put, done},
			loop(NewBuffer, NewObservers);

		{From, take, Size} ->
			{NewBuffer, NewObservers} = notify_observers(Buffer, Observers ++ [{from, From, size, Size, take}]),
			loop(NewBuffer, NewObservers);
		
		{From, peek, Size} ->
			{NewBuffer, NewObservers} = notify_observers(Buffer, Observers ++ [{from, From, size, Size, peek}]),
			loop(NewBuffer, NewObservers);
		
		{From, close} ->
			clean_observers(Observers),
			From ! {self(), close, done},
			erlang:exit(normal)
	end.

% no observers to notify
notify_observers(Buffer, []) ->
	{Buffer, []};
% found observer to taking data, and has sufficient data
notify_observers(Buffer, [{from, From, size, Size, take} | RestObservers])
  when erlang:byte_size(Buffer) >= Size ->
	<<Data:Size/binary, NewBuffer/binary>> = Buffer,
	From ! {self(), take, Data},
	notify_observers(NewBuffer, RestObservers);
% found observer for peeking data, and has sufficient data
notify_observers(Buffer, [{from, From, size, Size, peek} | RestObservers])
  when erlang:byte_size(Buffer) >= Size ->
	From ! {self(), peek, binary_part(Buffer, 0, Size)},
	notify_observers(Buffer, RestObservers);
% found observers, but buffer insufficient
notify_observers(Buffer, Observers) ->
	{Buffer, Observers}.

clean_observers([]) ->
	ok;
clean_observers([{from, From, size, _Size} | RestObservers]) ->
	From ! {self(), closed},
	clean_observers(RestObservers).

close(BlockingBinaryQueue) ->
	BlockingBinaryQueue ! {self(), close},
	receive
		{BlockingBinaryQueue, close, done} -> ok
	end.

put(BlockingBinaryQueue, Data) when is_binary(Data) ->
	BlockingBinaryQueue ! {self(), put, Data},
	receive
		{BlockingBinaryQueue, put, done} -> ok
	end;
put(_BlockingBinaryQueue, _Data) ->
	erlang:error(badarg).

% get first [size] bytes of data, and remove them from queue
take(BlockingBinaryQueue, Size) when erlang:is_integer(Size), Size > 0 ->
	BlockingBinaryQueue ! {self(), take, Size},
	receive
		{BlockingBinaryQueue, take, Data} ->
			Data;
		{BlockingBinaryQueue, closed} -> 
			{error, closed}
	end;
take(_BlockingBinaryQueue, _Size) ->
	erlang:error(badarg).

% get first [size] bytes of data, but don't remove them from queue
peek(BlockingBinaryQueue, Size) when erlang:is_integer(Size), Size > 0 ->
	BlockingBinaryQueue ! {self(), peek, Size},
	receive
		{BlockingBinaryQueue, peek, Data} ->
			Data;
		{BlockingBinaryQueue, closed} -> 
			{error, closed}
	end;
peek(_BlockingBinaryQueue, _Size) ->
	erlang:error(badarg).

%%
%% Local Functions
%%

